package com.wyx.springbootspringkafkademo.consumer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wyx.springbootspringkafkademo.entity.Message;
import com.wyx.springbootspringkafkademo.entity.User;
import com.wyx.springbootspringkafkademo.producer.KafkaProducerService;
import com.wyx.springbootspringkafkademo.producer.UserProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.generic.GenericRecord;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

/**
 * ClassName UserConsumer
 * Description 基于avro 反序列化
 *
 * @author qq
 * @date 2025/4/25 16:20
 * @since JDK 1.8
 */
@Component
@Slf4j
public class UserConsumer {

    @KafkaListener(topics = UserProducer.TOPIC, containerFactory ="avroKafkaListenerContainerFactory")
    public void consume(User record, Acknowledgment acknowledgment) {
        log.info("Received usr: {}", record);
        // 提交offset
        acknowledgment.acknowledge();
    }

//    @KafkaHandler
//    public void consume(@Payload  User user) {
//        System.out.println("Received user: " + user.getName());
//    }
}
